package io.kiki.sba.registry.server.metadata.slot.util.selector;

import io.kiki.sba.registry.common.model.slot.DataStoreNode;
import io.kiki.sba.registry.server.metadata.slot.util.builder.SlotsBuilder;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Set;

class DefaultSlotLeaderSelector implements Selector<String> {

    private final SlotsBuilder slotsBuilder;
    private final int highWaterMarkSlotLeaderNums;
    private final int slotId;

    public DefaultSlotLeaderSelector(int highWaterMarkSlotLeaderNums, SlotsBuilder slotsBuilder, int leaderSlotId) {
        this.highWaterMarkSlotLeaderNums = highWaterMarkSlotLeaderNums;
        this.slotsBuilder = slotsBuilder;
        this.slotId = leaderSlotId;
    }

    @Override
    public String select(Collection<String> dataStoreNodeIdList) {
        Set<String> followerDataStoreNodeIdSet = slotsBuilder.getOrCreate(slotId).getFollowerDataStoreNodeIdSet();
        Collection<String> followerDataStoreNodeIdList = new ArrayList<>(dataStoreNodeIdList);
        followerDataStoreNodeIdList.retainAll(followerDataStoreNodeIdSet);
        // first, try to select the candidate which is the follower
        String leader = new LeastLeaderFirstSelector(slotsBuilder).select(followerDataStoreNodeIdList);
        if (leader != null) {
            // check the num of leaders
            DataStoreNode dataStoreNode = slotsBuilder.getDataStoreNode(leader);
            if (dataStoreNode.getLeaderSlotIdSet().size() < highWaterMarkSlotLeaderNums) {
                return leader;
            }
        }
        // second, find other candidate
        return new LeastLeaderFirstSelector(slotsBuilder).select(dataStoreNodeIdList);
    }
}
